package kafka

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/IBM/sarama"
)

// StatusMessage represents a job status message
type StatusMessage struct {
	JobID      string                 `json:"job_id"`
	ClusterID  string                 `json:"cluster_id"`
	Status     string                 `json:"status"`
	Message    string                 `json:"message"`
	Details    map[string]interface{} `json:"details"`
	Timestamp  time.Time              `json:"timestamp"`
	RetryCount int                    `json:"retry_count"`
}

// Writer represents a Kafka message writer
type Writer struct {
	client    *Client
	producer  sarama.SyncProducer
	errorChan chan error
}

// NewWriter creates a new Kafka writer
func NewWriter(client *Client) (*Writer, error) {
	if client == nil {
		return nil, fmt.Errorf("client cannot be nil")
	}

	return &Writer{
		client:    client,
		producer:  client.GetProducer(),
		errorChan: make(chan error, 100),
	}, nil
}

// WriteMessage writes a message to Kafka
func (w *Writer) WriteMessage(ctx context.Context, topic string, key []byte, value []byte) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	msg := &sarama.ProducerMessage{
		Topic: topic,
		Key:   sarama.ByteEncoder(key),
		Value: sarama.ByteEncoder(value),
	}

	partition, offset, err := w.producer.SendMessage(msg)
	if err != nil {
		return fmt.Errorf("failed to send message to topic %s: %w", topic, err)
	}

	// Log successful write
	fmt.Printf("Message sent to topic %s, partition %d, offset %d\n", topic, partition, offset)
	return nil
}

// WriteJobStatus writes a job status message to Kafka
func (w *Writer) WriteJobStatus(ctx context.Context, statusMsg *StatusMessage) error {
	if statusMsg == nil {
		return fmt.Errorf("status message cannot be nil")
	}

	// Validate status message
	if err := ValidateStatusMessage(statusMsg); err != nil {
		return fmt.Errorf("invalid status message: %w", err)
	}

	// Serialize status message
	data, err := SerializeStatusMessage(statusMsg)
	if err != nil {
		return fmt.Errorf("failed to serialize status message: %w", err)
	}

	// Write to status topic
	statusTopic := w.client.config.Topic + "-status"
	key := []byte(statusMsg.JobID)

	return w.WriteMessage(ctx, statusTopic, key, data)
}

// WriteJobMessage writes a job message to Kafka
func (w *Writer) WriteJobMessage(ctx context.Context, jobMsg *JobMessage) error {
	if jobMsg == nil {
		return fmt.Errorf("job message cannot be nil")
	}

	// Validate job message
	if err := ValidateJobMessage(jobMsg); err != nil {
		return fmt.Errorf("invalid job message: %w", err)
	}

	// Serialize job message
	data, err := SerializeJobMessage(jobMsg)
	if err != nil {
		return fmt.Errorf("failed to serialize job message: %w", err)
	}

	// Write to job topic
	key := []byte(jobMsg.JobID)

	return w.WriteMessage(ctx, w.client.config.Topic, key, data)
}

// WriteBatch writes multiple messages to Kafka in a batch
func (w *Writer) WriteBatch(ctx context.Context, messages []*sarama.ProducerMessage) error {
	if len(messages) == 0 {
		return nil
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	err := w.producer.SendMessages(messages)
	if err != nil {
		return fmt.Errorf("failed to send batch messages: %w", err)
	}

	fmt.Printf("Successfully sent %d messages in batch\n", len(messages))
	return nil
}

// SerializeStatusMessage serializes a StatusMessage into bytes
func SerializeStatusMessage(statusMsg *StatusMessage) ([]byte, error) {
	if statusMsg == nil {
		return nil, fmt.Errorf("status message cannot be nil")
	}

	data, err := json.Marshal(statusMsg)
	if err != nil {
		return nil, fmt.Errorf("failed to serialize status message: %w", err)
	}

	return data, nil
}

// DeserializeStatusMessage deserializes bytes into a StatusMessage
func DeserializeStatusMessage(data []byte) (*StatusMessage, error) {
	if data == nil {
		return nil, fmt.Errorf("data cannot be nil")
	}

	var statusMsg StatusMessage
	if err := json.Unmarshal(data, &statusMsg); err != nil {
		return nil, fmt.Errorf("failed to deserialize status message: %w", err)
	}

	return &statusMsg, nil
}

// ValidateStatusMessage validates a StatusMessage
func ValidateStatusMessage(statusMsg *StatusMessage) error {
	if statusMsg == nil {
		return fmt.Errorf("status message cannot be nil")
	}

	if statusMsg.JobID == "" {
		return fmt.Errorf("job_id is required")
	}

	if statusMsg.ClusterID == "" {
		return fmt.Errorf("cluster_id is required")
	}

	if statusMsg.Status == "" {
		return fmt.Errorf("status is required")
	}

	// Validate status values
	validStatuses := map[string]bool{
		"pending":   true,
		"running":   true,
		"completed": true,
		"failed":    true,
		"cancelled": true,
	}

	if !validStatuses[statusMsg.Status] {
		return fmt.Errorf("invalid status: %s", statusMsg.Status)
	}

	return nil
}

// CreateStatusMessage creates a new StatusMessage
func CreateStatusMessage(jobID, clusterID, status, message string, details map[string]interface{}) *StatusMessage {
	return &StatusMessage{
		JobID:      jobID,
		ClusterID:  clusterID,
		Status:     status,
		Message:    message,
		Details:    details,
		Timestamp:  time.Now(),
		RetryCount: 0,
	}
}

// UpdateStatusMessage updates an existing StatusMessage
func UpdateStatusMessage(statusMsg *StatusMessage, status, message string, details map[string]interface{}) {
	if statusMsg == nil {
		return
	}

	statusMsg.Status = status
	statusMsg.Message = message
	if details != nil {
		statusMsg.Details = details
	}
	statusMsg.Timestamp = time.Now()
}

// Errors returns the error channel
func (w *Writer) Errors() <-chan error {
	return w.errorChan
}

// Close closes the writer
func (w *Writer) Close() error {
	// The producer is managed by the client, so we don't close it here
	return nil
}
